package com.fwmagic.spark.core.transformations

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 对两个RDD中的KV元素,每个RDD中相同key中的元素分别聚合成一个集合。
  * 与reduceByKey不同的是针对两个RDD中相同的key的元素进行合并。
  * 例子中将多个RDD中同一个Key对应的Value组合到一起。
  * rdd1中不存在Key为dd的元素(自然就不存在Value了),在组合的过程中将rdd1对应的位置
  * 设置为CompactBuffer()了,而不是去掉了。
  */
object CoGroupDemo {

    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf()
                .setAppName(this.getClass.getSimpleName)
                .setMaster("local[*]")

        val sc: SparkContext = new SparkContext(conf)

        val rdd1: RDD[(String, Int)] = sc.parallelize(Array(("aa", 1), ("bb", 2), ("cc", 6)))

        val rdd2: RDD[(String, Int)] = sc.parallelize(Array(("aa", 3), ("bb", 4), ("cc", 5), ("dd", 6), ("aa", 8)))

        val res: RDD[(String, (Iterable[Int], Iterable[Int]))] = rdd1.cogroup(rdd2)

        val res2: RDD[(String, Int)] = res.mapValues(tp => {
            tp._1.sum + tp._2.sum
        })

        /*
            (cc,11)
            (bb,6)
            (aa,12)
            (dd,6)
         */
        res2.collect().foreach(println)

        /*
            (cc,(CompactBuffer(6),CompactBuffer(5)))
            (bb,(CompactBuffer(2),CompactBuffer(4)))
            (aa,(CompactBuffer(1),CompactBuffer(3, 8)))
            (dd,(CompactBuffer(),CompactBuffer(6)))
         */
        //res.collect().foreach(println(_))

        sc.stop()
    }
}
